package mq

import (
	"fmt"
	"github.com/streadway/amqp"
	"logstash/config"
)

type RmqProducer struct {
	QueueName    string // 队列名称
	RoutingKey   string // key名称
	ExchangeName string // 交换机名称
	ExchangeType string // 交换机类型
}

func (p *RmqProducer) Send(msg string) error {
	// 验证链接是否正常,否则重新链接
	conn, err := p.connect()
	if err != nil {
		return err
	}
	ch, err := conn.Channel()
	if err != nil {
		fmt.Printf("MQ打开管道失败:%s \n", err)
		return err
	}
	defer func() {
		// 先关闭管道,再关闭链接
		err := ch.Close()
		if err != nil {
			fmt.Printf("MQ管道关闭失败 1:%s \n", err)
		}
		err = conn.Close()
		if err != nil {
			fmt.Printf("MQ链接关闭失败 1:%s \n", err)
		}
	}()
	//用于检查队列是否存在,已经存在不需要重复声明
	q, err := ch.QueueDeclarePassive(p.QueueName, false, false, false, true, nil)
	if err != nil {
		fmt.Printf("QueueDeclarePassive失败:%s \n", err)
		// 队列不存在,声明队列
		// name:队列名称;durable:是否持久化,队列存盘,true服务重启后信息不会丢失,影响性能;autoDelete:是否自动删除;noWait:是否非阻塞,
		// true为是,不等待RMQ返回信息;args:参数,传nil即可;exclusive:是否设置排他
		q, err = ch.QueueDeclare(p.QueueName, false, false, false, true, nil)
		if err != nil {
			fmt.Printf("MQ注册队列失败:%s \n", err)
			panic(err)
		}
	}
	err = ch.Publish(
		"",     // exchange
		q.Name, // routing key
		false,  // mandatory
		false,  // immediate
		amqp.Publishing{
			ContentType: "text/plain",
			Body:        []byte(msg),
		})
	//utile.FailOnError(err, "Failed to publish a message")
	if err != nil {
		fmt.Printf("Failed to publish a message:%s \n", err)
		return err
	}
	return nil
}

func (p *RmqProducer) connect() (*amqp.Connection, error) {
	conf := config.GetConfig()
	mqconf := conf.Data.Mq
	mqUrl := fmt.Sprintf("amqp://%s:%s@%s:%s/", mqconf.Username, mqconf.Password, mqconf.Host, mqconf.Port)
	conn, err := amqp.Dial(mqUrl)
	if err != nil {
		fmt.Printf("MQ打开链接失败:%s \n", err)
		return nil, err
	}
	return conn, nil
}

func (p *RmqProducer) getName() string {
	return p.QueueName
}
